﻿//-----------------------------------------------------------------------
// <copyright file="StreamObserver.cs" author="Artak Mkrtchyan">
//     Creative Commons - GNU General Public License (http://creativecommons.org/licenses/GPL/2.0/
// </copyright>
// <author>Artak Mkrtchyan</author>
// <email>mkArtak@gmail.com</email>
// <date>15/07/2012</date>
//-----------------------------------------------------------------------
using ConsoleHoster.Common.Utilities;
using System;
using System.Collections.Concurrent;
using System.IO;
using System.Text;
using System.Threading;

namespace ConsoleHoster.Model
{
	public sealed class StreamObserver : IDisposable
	{
		private readonly ILogger logger;
		private readonly StreamReader stream;
		private readonly StringBuilder messageBuffer = new StringBuilder();
		private readonly ConcurrentQueue<StreamMessage> messageQueue = new ConcurrentQueue<StreamMessage>();
		private Thread streamReadingThread;

		private bool isDisposed = false;
		private readonly Object disposalLock = new Object();

		public StreamObserver(StreamReader argStream, ILogger argLogger)
		{
			ValidationManager.RequireArgumentNotNull(argStream, "argStream");

			this.logger = new ComponentLogger("StreamObserver", argLogger);
			this.stream = argStream;
		}

		public void BeginObservation()
		{
			if (this.streamReadingThread != null)
			{
				throw new Exception("Already observing the stream");
			}

			this.streamReadingThread = new Thread(new ThreadStart(this.ObserveStreamInBackground));
			this.streamReadingThread.IsBackground = true;
			this.streamReadingThread.Priority = ThreadPriority.BelowNormal;
			this.streamReadingThread.Start();
		}

		public StreamMessage? GetNextAvailableMessage()
		{
			StreamMessage? tmpResult = null;

			StreamMessage tmpMessage;
			if (this.messageQueue.TryDequeue(out tmpMessage))
			{
				tmpResult = tmpMessage;
			}

			return tmpResult;
		}

		public void Dispose()
		{
			this.Dispose(true);
		}

		private void ObserveStreamInBackground()
		{
			try
			{
				DateTime tmpReceiveDate = DateTime.UtcNow;
				char? tmpStreamChar = null;

				do
				{
					int tmpStatus = this.stream.Peek();

					if (tmpStatus == -1)
					{
						if (this.messageBuffer.Length > 0)
						{
							this.InformMessage(tmpReceiveDate);
						}

						tmpStreamChar = (char)this.stream.Read();
					}

					char tmpCurrentChar = tmpStreamChar ?? (char)this.stream.Read();
					if (tmpStreamChar.HasValue)
					{
						tmpStreamChar = null;
					}

					if (this.messageBuffer.Length == 0)
					{
						tmpReceiveDate = DateTime.UtcNow;
					}

					this.messageBuffer.Append(tmpCurrentChar);
					if (tmpCurrentChar == '\n')
					{
						this.InformMessage(tmpReceiveDate);
					}
				} while (true);
			}
			catch (ThreadAbortException)
			{
				this.logger.LogMessage("Stream Observer thread aborted");
			}
			catch (Exception ex)
			{
				this.logger.LogError("Error while observing stream. Details: {0}", ex.ToString());
			}
		}

		private void InformMessage(DateTime argReceivedDate)
		{
			string tmpMessage = this.messageBuffer.ToString();
			this.messageBuffer.Clear();
			this.messageQueue.Enqueue(new StreamMessage(argReceivedDate, tmpMessage));
		}

		private void Dispose(bool argDisposing)
		{
			if (!this.isDisposed)
			{
				new Action(() =>
				{
					if (!this.isDisposed)
					{
						if (argDisposing)
						{
							try
							{
								this.stream.Close();
								this.streamReadingThread.Abort();
							}
							catch
							{
							}
							finally
							{
								this.streamReadingThread = null;
							}
						}

						this.isDisposed = true;
					}
				}).SyncronizeCallByLocking(this.disposalLock);
			}
		}
	}
}